package com.edu.yx.app.App_03_DwdApp.Dwd_02_User;

import com.edu.yx.app.APP_01_BaseApp.BaseSqlApp;
import com.edu.yx.common.Constant;
import com.edu.yx.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_01_User_ActDetail extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_01_User_ActDetail().init(
                3021,
                2,
                "Dwd_01_User_ActDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        //1.读取ods_db
        readOdsDb(tEnv,"Dwd_01_User_ActDetail");

        //2.从ods_db读取表user_info
        //todo 由于页面信息没有用户登陆行为，所以这里将以下两点
        // 1.用户的注册      （计算新增用户） type=insert
        // 2.用户对自身的更新 （计算活跃用户）  type=update

        Table result = tEnv.sqlQuery("select " +
                " cast(data['id'] as string) user_id," +
                " type," +
                " ts," +
                " pt " +
                " from ods_db" +
                " where `database` = 'edu' " +
                " and `table` = 'user_info' " +
                " and (`type` = 'insert') or (`type` = 'update' and `old` is not null ) "
        );

        tEnv.executeSql("create table dwd_user_act_detail(" +
                " user_id string, " +
                " type  string," +
                " ts bigint," +
                " pt TIMESTAMP_LTZ(3)" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_USER_ACT_DETAIL)
        );

        result.executeInsert("dwd_user_act_detail");

    }
}
